package com.dmp.storm.bolt;

import com.dmp.common.string.StringUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


/**
 * 
 * @author hutao
 *
 */
public class InputDataKeysSplit extends BaseMapBolt {

	/**
	 * 
	 */
	private static final long serialVersionUID = 101782063161347686L;

	private String inputKeys;// 输入key
	private String splitStr;// 分割符

	private String extraKeys;// 需要额外处理的字段
	private Map<String, Object> extraKeysMap;// 额外处理的字段

	public String getSplitStr() {
		return splitStr;
	}

	public void setSplitStr(String splitStr) {
		this.splitStr = splitStr;
	}

	public String getInputKeys() {
		return inputKeys;
	}

	public void setInputKeys(String inputKeys) {
		this.inputKeys = inputKeys;
	}

	public String getExtraKeys() {
		return extraKeys;
	}

	public void setExtraKeys(String extraKeys) {
		this.extraKeys = extraKeys;
	}

	@Override
	public Map<String, Object> process(Map<String, Object> map) {
		System.out.println(map);
		// TODO Auto-generated method stub
		try {
			String kafkamsg = (new String((byte[]) map.get("kafkamsg")));
			// System.err.println(kafkamsg);
			map.put("result", kafkamsg);
			String[] values = kafkamsg.split("\\|", -1);
			String[] keys = inputKeys.split("\\" + splitStr);
			for (int n = 0; n < keys.length; n++) {
				// map.put(keys[n].trim(), values[n].trim());
				String key = keys[n].trim();
				String value = values[n].trim();
				if (extraKeysMap.containsKey(key) && StringUtils.isBlank(value)) {
					map.put(key, extraKeysMap.get(key));
				} else {
					map.put(key, value);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return map;
	}

	@SuppressWarnings("rawtypes")
	@Override
	public void prepare() {
		try {
			extraKeysMap = new ConcurrentHashMap<>();
			System.out.println("extraKeys =======1111======= " + extraKeys);
			if (StringUtils.isNotBlank(extraKeys)) {
				System.out.println("extraKeys =====22222========= " + extraKeys);
				String[] extraKeysArray = extraKeys.split(",");
				for (int i = 0; i < extraKeysArray.length; i++) {
					String keyValue = extraKeysArray[i];
					String[] array = keyValue.split("=");
					String key = array[0];
					String value = array[1];
					extraKeysMap.put(key, value);
				}
			}
		} catch (Exception e) {
			System.err.println("解析extraKeys字段出错！");
			System.err.println(e.getMessage());
		}
	}

}
